package com.cheng.im.gateway.mq;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import com.cheng.im.common.bean.Notification;
import com.cheng.im.common.bean.UserDTO;
import com.cheng.im.common.bean.msg.ProtoMsgV1;
import com.cheng.im.common.constants.MQConstants;
import com.cheng.im.gateway.entity.ImWorker;
import com.cheng.im.gateway.session.LocalSession;
import com.cheng.im.gateway.session.ServerSession;
import com.cheng.im.gateway.session.dao.impl.RedissonUserCacheImpl;
import com.cheng.im.gateway.session.manager.SessionManger;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Service
public class RocketMQListen{

    @Resource
    SessionManger sessionManger;
    @Resource
    private RedissonUserCacheImpl userCacheDAO;


    @Value("${rocketmq.name-server}")
    private String namesrvAddr;

    public void initConsumers() {
        String convered = converString(ImWorker.getInst().getLocalNode().getBrokerId());


        DefaultMQPushConsumer consumer = MqFactory.getMQPushConsumer(namesrvAddr, MQConstants.SERVICE_2_IM +
                convered, "IM-CONSUMER-GROUP_" + convered);


        DefaultMQPushConsumer consumerIM = MqFactory.getMQPushConsumer(namesrvAddr, MQConstants.IM_2_IM +
                convered, "IM-CONSUMER-IM-GROUP_" + convered);

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 处理消息
                for (MessageExt msg : msgs) {
                    try {
//                        log.info("消费消息");
                        byte[] body = msg.getBody();
                        ProtoMsgV1.Message pkg = ProtoMsgV1.Message.parseFrom(body);
                        ProtoMsgV1.HeadType headType = pkg.getType();
                        log.info("消费类型：{}", headType.name());

                        if (headType.equals(ProtoMsgV1.HeadType.MESSAGE_REQUEST)) {
                            String sessionId = pkg.getSessionId();
                            // 单聊的转发
                            ServerSession session = SessionManger.inst().getSessionBySession(sessionId);
                            if (!ObjectUtil.isNull(session)) {
                                session.writeAndFlush(pkg);
                            } else {
                                // 清理缓存
                                userCacheDAO.removeSession(pkg.getMessageRequest().getTo(), sessionId, pkg.getAppId());
                            }
                        }else {
                            String sessionId = pkg.getSessionId();
                            ServerSession session = SessionManger.inst().getSessionBySession(sessionId);
                            if (!ObjectUtil.isNull(session)) {
                                session.writeAndFlush(pkg);
                            } else {
                                // 清理session，下面方法无法清理
                                // SessionManger.inst().removeSession(sessionId, pkg.getAppId());
                            }
                        }
                    } catch (Exception e) {
                        log.error("MQ消息监听的消息出错了");
                        // 重试消费
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumerIM.registerMessageListener(new MessageListenerConcurrently(){
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 处理消息
                for (MessageExt msg : msgs) {
                    try {
//                        log.info("消费消息");
                        byte[] body = msg.getBody();
                        ProtoMsgV1.Message pkg = ProtoMsgV1.Message.parseFrom(body);
                        ProtoMsgV1.HeadType headType = pkg.getType();
                        log.info("消费类型：{}", headType.name());

                        if (headType.equals(ProtoMsgV1.HeadType.MESSAGE_NOTIFICATION)) {

                            // 接收通知
                            String extra = pkg.getNotification().getExtra();
                            Notification notification = JSONObject.parseObject(extra, Notification.class);
                            if (notification.getType() == Notification.SESSION_OFF) {
                                // 强制下线
                                String sessionId = pkg.getSessionId();
                                LocalSession sessionBySession = (LocalSession) sessionManger.getSessionBySession(sessionId);
                                if (ObjectUtil.isNotNull(sessionBySession)) {
                                    sessionBySession.close();
                                    // 删除本地的会话和远程会话
                                    // sessionManger.removeSession(sessionId, pkg.getAppId());
                                } else {
                                    // 清理过期的远程会话
                                    Notification<UserDTO> notification1 = JSON.parseObject(
                                            extra,
                                            new TypeReference<Notification<UserDTO>>() {}
                                    );
                                    UserDTO userDTO = notification1.getData();
                                    userCacheDAO.removeSession(userDTO.getUserId(), userDTO.getSessionId(), userDTO.getAppId());
                                }
                            }
                        }
                    } catch (Exception e) {
                        log.info("MQ消息监听的消息出错了");
                        // 重试消费
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        } );


        try {
            consumer.start();
            consumerIM.start();
        } catch (MQClientException e) {
            log.error("==============消息监听启动出错================");
            throw new RuntimeException(e);
        }
        log.info("mq初始化监听节点成功");
    }

    @PreDestroy
    public void destroy(){
        MqFactory.destroy();
    }

    private String converString(String source){
        return source.replace(".", "-").replace(":", "_");
    }
}
